package transport.handler;

import hdfs.TransPort;
import io.netty.channel.*;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import transport.ChannelUtils;
import transport.ChannelContext;

/**
 * data transfer between channels
 */
public final class InnerHandler extends ChannelInboundHandlerAdapter {
    private static final Logger LOG = LoggerFactory.getLogger(InnerHandler.class);
    private Channel relayChannel = null;
    private final ChannelContext relayCtx;

    public InnerHandler(ChannelContext relayCtx) {
        this.relayCtx = relayCtx;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        ChannelUtils.tryToReadIfNeeded(ctx);
//        relayChannel = MyChannel.getRelayChannel();
//        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
//        TransPort.WritePacketResponse response = (TransPort.WritePacketResponse) msg;
        TransPort.WritePacketRequest response = (TransPort.WritePacketRequest) msg;
        if(relayChannel==null){
            relayChannel = relayCtx.getRelayChannel();
        }
        if (relayChannel.isActive()) {
            LOG.debug("{} 获取到回复状态 {} {}",ctx.channel().localAddress(),response.getSize(),relayChannel.isWritable());
            relayChannel.writeAndFlush(response).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (channelFuture.isSuccess()) {
                        ChannelUtils.tryToReadIfNeeded(ctx);
                    } else {
                        ChannelUtils.closeOnFlush(relayChannel);
                    }
                }
            });
        } else {
            ReferenceCountUtil.release(msg);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        LOG.info("channel {} closed", ctx.channel());
        ChannelUtils.closeOnFlush(relayChannel);
        ctx.close();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        LOG.error("data transfer handler occurs error, channel: {}", ctx.channel(), cause);
        ctx.close();
    }

}
